package org.example

import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern

object ketang3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext
    // 电影永恒分析数据概况
    val filePath="src/main/resources/"
    val moviesRDD=sc.textFile(filePath +"movies.dat")
    val occupationsRDD=sc.textFile(filePath +"occupations.dat")
    val ratingsRDD=sc.textFile(filePath +"ratings.dat")
    val usersRDD=sc.textFile(filePath +"users.dat")
    println("总电影数"+moviesRDD.count())
    println("总用户数"+usersRDD.count())
    println("评分条数"+ratingsRDD.count())
    println("职业数"+occupationsRDD.count())
    /*//一、用户分析：统计某用户评过分的电影数量并打印电影详情
    val userRating=ratingsRDD.map(_.split("::"))
      .map(user =>(user(1),user(0)))//转换为key（电影id）value（电影id）
     // .foreach(println)  直接打印结果
     .filter(_._2.equals("2688"))  //统计用户id为2688的影评数量
    println("用户id为2688的影评数量为："+userRating.count())

    val movieInfo =moviesRDD.map(_.split("::"))
      .map(movie =>(movie(0),(movie(1),movie(2))))//将电影数据转换为key value型
      .join(userRating)//join连接后只剩下两表都有key（2688）的电影数据
      .map(item => (item._1,item._2._2,item._2._1,item._2._1._2))
    //转换成一维的数据格式(电影id，用户id，电影名，电影类型）
    //movieInfo.foreach(println)//全部打印
    movieInfo.take(10).foreach(println)//打印前十个*/


    /*//二、电影分析：找出评分最高的前十部电影和观看次数最多的前十部电影
    val ratings = ratingsRDD.map(_.split("::"))
      .map(rating => (rating(1), (rating(2).toDouble, 1))) // (电影ID, (评分, 1))
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)) // (电影ID, (总评分, 评分次数))
      .mapValues { case (totalScore, count) => totalScore / count } // (电影ID, 平均评分)
      .sortBy(_._2, ascending = false) // 按平均评分降序排序
      .take(10) // 取前10部

    println("评分最高的前十部电影：")
    ratings.foreach { case (movieId, avgScore) =>
      val movieDetails = moviesRDD.filter(line => line.split("::")(0) == movieId).first().split("::")
      println(s"电影ID: $movieId, 平均评分: $avgScore, 电影名: ${movieDetails(1)}, 类型: ${movieDetails(2)}")
    }
    // 观看次数最多的十部电影
    val watchCounts = ratingsRDD.map(_.split("::"))
      .map(rating => (rating(1), 1)) // (电影ID, 1)
      .reduceByKey(_ + _) // (电影ID, 观看次数)
      .sortBy(_._2, ascending = false) // 按观看次数降序排序
      .take(10) // 取前10部

    println("观看次数最多的十部电影：")
    watchCounts.foreach { case (movieId, count) =>
      val movieDetails = moviesRDD.filter(line => line.split("::")(0) == movieId).first().split("::")
      println(s"电影ID: $movieId, 观看次数: $count, 电影名: ${movieDetails(1)}, 类型: ${movieDetails(2)}")
    }*/


    /*val rating = ratingsRDD.map(x => x.split("::")).map{
      x =>{
        (x(0), x(1), x(2))
      }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
    rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x,y) =>{
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x, y) =>{
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
      }).sortByKey(false)
      .take(5)
      .foreach(println)
    println("观影人数最多的前 10 部电影")
    val watchViewsInfo = rating.map(x =>{
      (x._2, 1)
    }).reduceByKey((x, y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)
    watchViewsInfo.foreach(println(_))
    println("===================>")
    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) =>{
        (x + y)
      })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)
    println("详情的输出(  观影人数，电影编号)")
    moviesInfo.join(ratingsInfo).map(x => {
      (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
    }).sortByKey(false)
      .take(10)
      .foreach(println)*/


    //三、电影类型统计
    val movieStyle=moviesRDD.map(_.split("::"))
      .map(x => (x(0),x(2)))//只留电影id和电影类型
      .flatMapValues(x =>x.split("\\|"))//根据符号拆分
      .map(x =>(x._2,1))//转为key value类型
    //在控制台打印科幻片的电影数量
      .reduceByKey((x,y)=>x+y)
      .filter(_._1.equals("Fantasy"))
    movieStyle.foreach(println)
    //movieStyle.take(10).foreach(println)

    //四、年度电影数量统计
    //正则表达式 找出小括号里的四位数
    val pattern=Pattern.compile("(.*)(\\(\\d{4}\\))")

    val movieYear=moviesRDD.map(_.split("::"))
      .map(x => (x(1),1))
      .map(x => {
        var name=""
        var year=""
        val matcher=pattern.matcher(x._1)
        if (matcher.find()) {
          name = matcher.group(1)
          year = matcher.group(2)
          year =year.substring(1,year.length-1)
        }
        if (year == ""){
          (-1,1)
        }else {
          (year.toInt,1)
        }
      })
      .reduceByKey((x,y)=>x+y)
    movieYear.foreach(println)
    val movies2000 = moviesRDD.filter(line => line.split("::")(1).contains("(2000)"))
    val movies2000Count = movies2000.count()

    // 2. 统计2000年度喜剧片的数量
    val comedyMovies2000 = movies2000.filter(line => line.split("::")(2).contains("Comedy"))
    val comedyMovies2000Count = comedyMovies2000.count()

    println(s"2000年度电影的生产数量: $movies2000Count")
    println(s"2000年度喜剧片的数量: $comedyMovies2000Count")
    sc.stop()
  }

}
